package com.atguigu.realtime.app.dws;

import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.DimQuery;
import com.atguigu.realtime.util.MyJDBCUtil;
import com.atguigu.realtime.util.MyJedisUtil;
import com.atguigu.realtime.util.MyThreadPooluUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.sql.DriverManager;
import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;
/**
 * @ClassName: DimAsyncFunc
 * @Description:
 * @Author: kele
 * @Date: 2021/4/26 21:04
 **/

/**
 * 通过异步查询的方式关联到维度表：  异步就是将连接放入连接池中，
 * 封装异步函数：
 * 1、获取线程池,获取连接
 * 2、查询数据
 *
 */
abstract public class DimAsyncFunc<T> extends RichAsyncFunction<T,T> {

    private ThreadPoolExecutor pool;
    private Connection connect;

    @Override
    public void open(Configuration parameters) throws Exception {

        pool = MyThreadPooluUtil.getPool();
        connect = MyJDBCUtil.getConnect(Constant.PHOENIX_URL, Constant.PHOENIX_DRIVER);

       // Class.forName(Constant.PHOENIX_DRIVER);
       // connect = DriverManager.getConnection(Constant.PHOENIX_URL);
    }

    protected abstract void addDim(T input, Connection connect, Jedis redis,ResultFuture resultFuture);


    @Override
    public void asyncInvoke(T input,  //input 进入线程池的数据，resultFuture将结果作为一个集合返回
                            ResultFuture<T> resultFuture) throws Exception {

        pool.execute(new Runnable() {
            @Override
            public void run() {

                Jedis redis = MyJedisUtil.getRedis();

                //要查询好多表的信息，不知道需要查询哪些表，所以需要将方法传出去
                addDim(input,connect,redis,resultFuture);
                redis.close();
            }
        });
    }

    @Override
    public void close() throws Exception {
        if (connect != null) {
            connect.close();
        }

        if (pool != null) {
            pool.shutdown();
        }
    }
}
